package com.leyantech.ecp.agenx.task.filter;

import com.leyantech.ecp.agenx.common.redis.RedissonOperation;
import com.leyantech.ecp.agenx.common.redis.RedissonOperation.RedisExecutor;
import com.leyantech.ecp.agenx.common.utils.JsonUtils;
import com.leyantech.utility.Logger;
import com.leyantech.utility.config.ConfigUtil;

import io.vertx.core.json.Json;
import org.apache.commons.lang3.ObjectUtils;
import org.redisson.api.RRateLimiter;
import org.redisson.api.RateIntervalUnit;
import org.redisson.api.RateLimiterConfig;
import org.redisson.api.RateType;

import java.time.Duration;
import java.util.Objects;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.function.Supplier;

/**
 * @author eahau, {@literal <eahau@foxmail.com>}
 * @date 2022-11-15.
 */
public abstract class RateLimiter<Task> implements TaskFilter<Task> {

  protected final Logger log = new Logger(getClass());

  final RedissonOperation<RRateLimiter> rateLimiterOperation = RedissonOperation
      .rateLimiter("task");

  final Supplier<RateLimiterConfig> rateLimiterCommonConfig = ConfigUtil.getDynamicConfig(
      "rateLimiter",
      "dispatch.common." + getClass().getSimpleName(),
      json -> Json.CODEC.fromString(json, RateLimiterConfig.class),
      new RateLimiterConfig(RateType.OVERALL, TimeUnit.SECONDS.toMillis(10), 30L)
  );

  protected abstract String[] getRaterKeys(Task task);

  protected RateLimiterConfig getRateLimiterConfig(Task task) {
    return rateLimiterCommonConfig.get();
  }

  @Override
  public boolean test(Task task) {
    final String key = String.join(".", getRaterKeys(task));
    final RateLimiterConfig config = getRateLimiterConfig(task);

    log.info("rateLimiter name {}, config {}", key, Json.CODEC.toString(config));

    if (!validRateLimiterConfig(config)) {
      log.warn("rateLimiter config invalid.");
      return false;
    }

    final long rateIntervalMills = config.getRateInterval();

    final RedisExecutor<RRateLimiter> redisExecutor = rateLimiterOperation.executor(key);

    if (!redisExecutor.get(RRateLimiter::isExists, "isExists")) {
      redisExecutor.run(rateLimiter ->
          rateLimiter.trySetRate(
              config.getRateType(),
              config.getRate(),
              rateIntervalMills,
              RateIntervalUnit.MILLISECONDS
          ), "trySetRate"
      );
    }

    final RateLimiterConfig limitConfigInRedis = redisExecutor
        .get(RRateLimiter::getConfig, "getConfig");
    // 更新限流配置
    if (!equalsRateLimiterConfig(config, limitConfigInRedis)) {
      log.info("update rate config, old {}", JsonUtils.lazyToString(limitConfigInRedis));
      redisExecutor.run(rateLimiter ->
          rateLimiter.setRate(
              config.getRateType(),
              config.getRate(),
              rateIntervalMills,
              RateIntervalUnit.MILLISECONDS
          ), "setRate"
      );
    }

    final int jitterMills = ThreadLocalRandom.current().nextInt(1000, 10000);

    redisExecutor.run(
        rateLimiter -> rateLimiter.expireAsync(Duration.ofMillis(rateIntervalMills + jitterMills)),
        "expireAsync");

    final boolean acquire = redisExecutor.get(RRateLimiter::tryAcquire, "tryAcquire");

    if (!acquire) {
      log.info("filter by RateLimiter");
    }
    return acquire;
  }

  private boolean equalsRateLimiterConfig(RateLimiterConfig r1, RateLimiterConfig r2) {
    return Objects.equals(r1.getRate(), r2.getRate())
        && Objects.equals(r1.getRateType(), r2.getRateType())
        && Objects.equals(r1.getRateInterval(), r2.getRateInterval());
  }

  private boolean validRateLimiterConfig(RateLimiterConfig config) {
    return config != null
        && ObjectUtils.allNotNull(config.getRate(), config.getRateType(), config.getRateInterval())
        && config.getRate() > 0 && config.getRateInterval() > 0;
  }

}
